package com.cennavi.funcs;

import java.text.DecimalFormat;
import java.util.HashMap;

import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月29日 下午1:05:20 
    * @version 1.0 
    */

public class DerivedFlow2SatPairFunction implements PairFunction<Tuple2<String, String>, String, String> {

	private static final long serialVersionUID = 1L;

	private HashMap<String,String> rtic_capaMap;
	public DerivedFlow2SatPairFunction(Broadcast<HashMap<String,String>> rtic_capaBroadMap){
		this.rtic_capaMap = rtic_capaBroadMap.value();
	}
	
	public Tuple2<String, String> call(Tuple2<String, String> arg0) throws Exception {
		String[] data = arg0._2.split(",");
		String key = data[5];
		String value = "";
		String capacity = "null";
		String saturation = "null";
		if(this.rtic_capaMap.containsKey(key)){
			capacity = this.rtic_capaMap.get(key);
			if(!capacity.equals("R3") && !capacity.equals("R4")){
				//一小时通行能力换算为15分钟
				capacity = String.valueOf((int)Math.ceil(Integer.parseInt(capacity) / 4.0));
			}
		}
		if(!capacity.equals("null") && !capacity.equals("R3") && !capacity.equals("R4")){
			saturation = new DecimalFormat("######0.00").format(
					1.0 * Integer.parseInt(data[9]) / Integer.parseInt(capacity));
		}
		//城市代号
		value += "610100,";
		//时间戳
		value += data[1] + ",";
		//计算时间间隔
		value += "2,";
		//流量空间粒度
		value += "2,";
		//获取方式
		value += "2,";
		//路段序号
		value += key + ",";
		//速度
		value += data[7] + ",";
		//交通指数
		value += data[8] + ",";
		//流量
		value += data[9] + ",";
		//饱和度
		value += saturation + ",";
		//通行能力
		value += capacity;
		return new Tuple2<String, String>(key, value);
	}
	

}
